package com.gitee.sparta.core.cluster.strategy;

import com.gitee.sparta.common.helpers.CollHelper;
import com.gitee.sparta.common.helpers.IsolationHelper;
import com.gitee.sparta.common.helpers.JsonHelper;
import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;

import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;

/**
 * <p>
 *      follower 连接上 leader，leader 给 follower 分配 id
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 */
@Slf4j
public class LeaderAcceptStrategy extends Strategy {

    public LeaderAcceptStrategy(RedissonClient client,
                                LoadBalance loadBalance,
                                SpartaCoreProperties properties) {
        super(client, loadBalance, properties);
    }

    @Override
    public void handle(ChannelHandlerContext ctx, SocketChannel sc, ClusterProtocol cp) {
        RLock lock = this.client.getLock(this.properties.getLock());
        try {
            lock.lock();
            RMap<Integer, ServerBean> serverTable = this.client.getMap(properties.getServerTable());
            // 连接上客户端的 sid
            final int cliSid = CollHelper.lackNum(serverTable.keySet());
            final String isolation = IsolationHelper.isolation(cliSid);
            // ClusterCache.leaderIsolation = isolation;

            ServerBean bean = new ServerBean().setIsolation(isolation)
                    .setLeader(false).setServerId(cliSid);

            // 注册服务到 redis
            serverTable.put(cliSid, bean);
            // leader 才会进行缓存全部  follower 只会换存自己
            ClusterCache.putServerInfo(cliSid, bean);

            ClusterProtocol pro = new ClusterProtocol(
                    ClusterCmd.ACCEPT_RESPONSE,
                    cliSid, ClusterCache.getServerId());

            if (ClusterCache.isNetty()) {
                ClusterCache.putClient(ctx.channel(), cliSid);
                pro.setChannel(ctx.channel());
                InetSocketAddress address = (InetSocketAddress)ctx.channel().remoteAddress();
                log.info("[Sparta] Channel accept, sid: {}, addr: {}, protocol: {}",
                        cliSid, address, JsonHelper.getJsonString(cp));
            } else {
                ClusterCache.putClient(sc, cliSid);
                pro.setCli(sc);
                log.info("[Sparta] Channel accept, sid: {}", cliSid);
            }

            this.loadBalance.addNode(isolation, cliSid);
            AbstractClusterLeader.sendQueue(pro);
        } finally {
            lock.unlock();
        }
    }
}
